package com.cshuig.hadoop.hbase;


import com.cshuig.hadoop.hbase.model.ColumnFamily;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;

/**
 * Hello world!
 *
 */
public class HbaseDemo {
    private static HBaseAdmin hBaseAdmin;
    private static Configuration configuration;

    private static void init() throws IOException {
        configuration = HBaseConfiguration.create();
        // 先连接 zookeeper
        configuration.set("hbase.zookeeper.quorum", "hadoopip-02:2181,hadoopip-03:2181,hadoopip-04:2181");
        hBaseAdmin = new HBaseAdmin(configuration);
    }
    public static void main( String[] args ) throws IOException {
        init();
        //ColumnFamily columnFamily = ColumnFamily.InstanceHolder.columnFamily.set("base", 2).set("info").set("data", 3);
        //createTable("sales", columnFamily.getList());

        HTable hTable = new HTable(configuration, "sales");
//        for (int i = 1; i <= 100; i++) {
//            putCell(hTable, "rk000" + i, "base", "name", "car" + i);
//            putCell(hTable, "rk000" + i, "info", "name", "land rover" + i);
//            putCell(hTable, "rk000" + i, "data", "name", "2000000" + i);
//        }

        //getRow(hTable, "rk0001");

        //scanAllRows(hTable);

        scanRange(hTable, "rk0001", "rk0003");

        //scanFilter()
    }

    public static void createTable(String tableName, List<ColumnFamily> columnFamilyList) throws IOException {
        if (hBaseAdmin.tableExists(tableName)) {
            System.out.println(String.format("表:%s 已存在", tableName));
        } else {
            HTableDescriptor hTableDescriptor = new HTableDescriptor(TableName.valueOf(tableName));
            for (ColumnFamily columnFamily : columnFamilyList) {
                HColumnDescriptor hColumnDescriptor = new HColumnDescriptor(columnFamily.getName());
                hColumnDescriptor.setMaxVersions(columnFamily.getVersions());
                hTableDescriptor.addFamily(hColumnDescriptor);
            }
            hBaseAdmin.createTable(hTableDescriptor);
            hBaseAdmin.close();
        }
    }

    public static void putCell(HTable hTable, String rowkey, String columnFamily, String qualifier, String data) throws IOException {
        Put put = new Put(Bytes.toBytes(rowkey));
        put.add(Bytes.toBytes(columnFamily), Bytes.toBytes(qualifier), Bytes.toBytes(data));
        hTable.put(put);
        hTable.close();
    }

    public static Result getRow(HTable hTable, String rowkey) throws IOException {
        Get get = new Get(Bytes.toBytes(rowkey));
        Result result = hTable.get(get);
        System.out.println(result);
        hTable.close();
        return result;
    }

    public static void deleteRow(HTable hTable, String rowkey) throws IOException {
        Delete delete = new Delete(Bytes.toBytes(rowkey));
        hTable.delete(delete);
        hTable.close();
    }

    public static ResultScanner scanAllRows(HTable hTable) throws IOException {
        Scan scan = new Scan();
        ResultScanner resultScanner = hTable.getScanner(scan);
        for (Result result : resultScanner) {
            System.out.println(result);
        }
        hTable.close();
        return resultScanner;
    }

    public static ResultScanner scanRange(HTable hTable, String startRow, String stopRow) throws IOException {
        Scan scan = new Scan(Bytes.toBytes(startRow), Bytes.toBytes(stopRow));
        ResultScanner resultScanner = hTable.getScanner(scan);
        for (Result result : resultScanner) {
            System.out.println(result);
        }
        hTable.close();
        return resultScanner;
    }

    public static ResultScanner scanFilter(HTable hTable, String startRow, Filter filter) throws IOException {
        Scan scan = new Scan(Bytes.toBytes(startRow), filter);
        ResultScanner resultScanner = hTable.getScanner(scan);
        for (Result result : resultScanner) {
            System.out.println(result);
        }
        hTable.close();
        return resultScanner;
    }
}
